package org.budo.elasticsearch.jdbc.driver.statement;

import java.sql.PreparedStatement;
import java.sql.SQLException;

import org.budo.elasticsearch.jdbc.driver.BudoElasticsearchJdbcConnection;
import org.budo.elasticsearch.jdbc.driver.BudoElasticsearchJdbcPreparedStatement;
import org.elasticsearch.action.delete.DeleteRequestBuilder;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.search.SearchHit;

/**
 * @author limingwei
 */
public class DeleteStatement extends AbstractElasticsearchStatement {
    @Override
    public void execute(PreparedStatement preparedStatement) throws SQLException {
        SearchHit[] searchHits = this.executeSearchRequest(preparedStatement, new String[] { "_id" });

        for (SearchHit searchHit : searchHits) {
            this.deleteById(preparedStatement, searchHit.getId());
        }

        ((BudoElasticsearchJdbcPreparedStatement) preparedStatement).setUpdateCount(searchHits.length);
    }

    private void deleteById(PreparedStatement preparedStatement, String id) throws SQLException {
        BudoElasticsearchJdbcConnection connection = (BudoElasticsearchJdbcConnection) preparedStatement.getConnection();
        TransportClient transportClient = connection.getTransportClient();

        DeleteRequestBuilder deleteRequestBuilder = transportClient.prepareDelete().setIndex(connection.getCatalog()) //
                .setType(this.tableName()) //
                .setId(id);

        this.executeAndGet(deleteRequestBuilder);
    }
}
